import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

class MyResource{
    private volatile boolean FLAG=true;
    AtomicInteger atomicInteger=new AtomicInteger();
    private BlockingQueue<String> blockingQueue=null;

    public MyResource(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    public void myProd() throws Exception {
        String data=null;
        while (FLAG){
            data = atomicInteger.getAndIncrement()+"";
            boolean offer = blockingQueue.offer(data,2L, TimeUnit.SECONDS);
            if(offer==true){
                System.out.println(Thread.currentThread().getName()+"\t 生产"+data+"成功！");
            }else {
                System.out.println(Thread.currentThread().getName()+"\t 生产"+data+"失败！");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName() + "\t 停止生产，表示FLAG=false，生产结束");
    }

    public void myConsumer() throws Exception{
        String result;
        while (FLAG){
            result = blockingQueue.poll(2L, TimeUnit.SECONDS);
            if(result!=null&&result!=""){
                System.out.println(Thread.currentThread().getName()+"\t"+"消费"+result+"成功！");
            }else {
                FLAG=false;
                System.out.println(Thread.currentThread().getName()+"\t"+"消费失败，队列已为空！");
                return;
            }

        }
    }

    /**
     * 停止生产的判断
     */
    public void stop() {
        this.FLAG = false;
    }
}

/**
 * @auther: Alessio
 * @date: 2021/1/14
 */
public class ProdConsumer_BlockQueueDemo {
    public static void main(String[] args) {

        MyResource myResource = new MyResource(new ArrayBlockingQueue(10));

        new Thread(()->{
            System.out.println(Thread.currentThread().getName() + "\t 生产线程启动");
            try {
                myResource.myProd();
            } catch (Exception e) {
                e.printStackTrace();
            }
        },"AA").start();

        new Thread(()->{
            System.out.println(Thread.currentThread().getName() + "\t 消费线程启动");
            try {
                myResource.myConsumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        },"BB").start();

        try {
            TimeUnit.SECONDS.sleep(5);
            myResource.stop();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
